package spark

import org.apache.spark.SparkConf
import org.apache.spark.sql.functions.{col, length, lit, when}
import org.apache.spark.sql.{SaveMode, SparkSession}

import java.util.Properties


object HiveConnective {
  // parameters
  LoggerUtil.setSparkLogLevels()

  def main(args: Array[String]): Unit = {
    //System.setProperty("hadoop.home.dir", "D:\\hadoop-2.7.4")

    val conf = new SparkConf()
      .setAppName(this.getClass.getSimpleName)
      .setMaster("local[*]")

    val session = SparkSession.builder()
      .config(conf)
      .getOrCreate()

    // 连接远程Hive数据库
    val reader = session.read.format("jdbc")
      .option("url", "jdbc:hive2://172.29.4.17:10000/default")
      .option("user", "student")
      .option("password", "nju2023")
      .option("driver", "org.apache.hive.jdbc.HiveDriver")

    val registerHiveDqlDialect = new RegisterHiveSqlDialect()
    registerHiveDqlDialect.register()

    // 要处理的表
    val handler = new ELTHandler()
    val writer = new ClickHouseWriter()
//    val tblNameDsts = List("pri_cust_liab_info", "pri_cust_liab_acct_info", "pri_cust_base_info",
//                            "pri_star_info", "dm_v_as_djk_info", "dm_v_as_djkfq_info",
//                            "pri_credit_info", "pri_cust_asset_acct_info", "pri_cust_asset_info")
    val tblNameDsts = List("pri_cust_asset_info")  // 用来调试某一个表
    // 处理每一个表
    for (tblNameDst <- tblNameDsts) {
      // 获取表的所有记录
      var df = reader.option("dbtable", tblNameDst).load()

      if(tblNameDst.equals("pri_cust_liab_info")){
        df = handler.handle_pri_cust_liab_info(df)
      }else if(tblNameDst.equals("pri_cust_liab_acct_info")){
        df = handler.handle_pri_cust_liab_acct_info(df)
      }else if(tblNameDst.equals("pri_cust_base_info")){
        df = handler.handle_pri_cust_base_info(df)
      }else if(tblNameDst.equals("pri_star_info")){
        df = handler.handle_pri_star_info(df)
      }else if(tblNameDst.equals("dm_v_as_djk_info")){
        df = handler.handle_dm_v_as_djk_info(df)
      }else if(tblNameDst.equals("dm_v_as_djkfq_info")){
        df = handler.handle_dm_v_as_djkfq_info(df)
      }else if(tblNameDst.equals("pri_credit_info")){
        df = handler.handle_pri_credit_info(df)
      }else if(tblNameDst.equals("pri_cust_asset_acct_info")){
        df = handler.handle_pri_cust_asset_acct_info(df)
      }else if(tblNameDst.equals("pri_cust_asset_info")){
        df = handler.handle_pri_cust_asset_info(df)
      }else if(tblNameDst.equals("pri_cust_contact_info")) {
        df = handler.handle_pri_cust_contact_info(df)
      }

      // 写入click数据库 (暂时不写入)
       writer.writeToClickHouse(tblNameDst, df)
    }

    // 结束
    session.close()
  }
}
